package com.gitee.ixtf.vertx.rabbitmq

import cn.hutool.core.util.StrUtil
import com.fasterxml.jackson.databind.JsonNode
import com.gitee.ixtf.core.MAPPER
import com.gitee.ixtf.guice.Jguice
import com.gitee.ixtf.guice.kotlinx.get
import com.gitee.ixtf.vertx.rabbitmq.guice.RabbitMQSender
import com.rabbitmq.client.AMQP.BasicProperties
import io.vertx.core.buffer.Buffer
import io.vertx.core.json.JsonArray
import io.vertx.core.json.JsonObject
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.reactor.asFlux
import org.reactivestreams.Publisher
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.rabbitmq.OutboundMessage

const val AMQ_QUEUE = StrUtil.EMPTY
const val AMQ_TOPIC = "amq.topic"
const val AMQ_DIRECT = "amq.direct"
const val AMQ_FANOUT = "amq.fanout"

suspend fun OutboundMessage.send(sender: RabbitMQSender = Jguice.get()) =
    Mono.just(this).send(sender)

suspend fun Publisher<OutboundMessage>.send(sender: RabbitMQSender = Jguice.get()) =
    sender.send(this)

suspend fun Iterable<OutboundMessage>.send(sender: RabbitMQSender = Jguice.get()) =
    Flux.fromIterable(this).send(sender)

suspend fun Flow<OutboundMessage>.send(sender: RabbitMQSender = Jguice.get()) =
    this.asFlux().send(sender)

fun message(
    exchange: String,
    routingKey: String,
    body: ByteArray,
    block: (BasicProperties.Builder.() -> Unit)? = null,
): OutboundMessage {
  val builder = BasicProperties.Builder().deliveryMode(2)
  block?.let { block(builder) }
  return OutboundMessage(exchange, routingKey, builder.build(), body)
}

fun topicMessage(
    routingKey: String,
    body: ByteArray,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = message(AMQ_TOPIC, routingKey, body, block)

fun topicMessage(
    routingKey: String,
    body: JsonNode,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = message(AMQ_TOPIC, routingKey, MAPPER.writeValueAsBytes(body), block)

fun topicMessage(
    routingKey: String,
    body: Buffer,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = topicMessage(routingKey, body.bytes, block)

fun topicMessage(
    routingKey: String,
    body: JsonObject,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = topicMessage(routingKey, body.toBuffer(), block)

fun topicMessage(
    routingKey: String,
    body: JsonArray,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = topicMessage(routingKey, body.toBuffer(), block)

fun topicMessage(
    routingKey: String,
    body: String,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = topicMessage(routingKey, Buffer.buffer(body), block)

fun queueMessage(
    queue: String,
    body: ByteArray,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = message(AMQ_QUEUE, queue, body, block)

fun queueMessage(
    routingKey: String,
    body: JsonNode,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = message(AMQ_TOPIC, routingKey, MAPPER.writeValueAsBytes(body), block)

fun queueMessage(
    queue: String,
    body: Buffer,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = queueMessage(queue, body.bytes, block)

fun queueMessage(
    queue: String,
    body: JsonObject,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = queueMessage(queue, body.toBuffer(), block)

fun queueMessage(
    queue: String,
    body: JsonArray,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = queueMessage(queue, body.toBuffer(), block)

fun queueMessage(
    queue: String,
    body: String,
    block: (BasicProperties.Builder.() -> Unit)? = null,
) = queueMessage(queue, Buffer.buffer(body), block)
